package streamAPI.checkpoint;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;

import java.io.IOException;
import java.nio.charset.Charset;

/**
 * @author yue.cao
 * @since 11-06-2020
 */
public class ErrorStringSchema implements DeserializationSchema<String>{

	@Override
	public String deserialize(byte[] message) throws IOException {
		String value = new String(message, Charset.forName("UTF-8"));
		if (value.equals("null")) {
			return null;
		} else if (value.equals("error")) {
			throw new RuntimeException("errorStringSchema deserialize error");
		}
		return value;
	}

	@Override
	public boolean isEndOfStream(String nextElement) {
		return false;
	}

	@Override
	public TypeInformation<String> getProducedType() {
		return BasicTypeInfo.STRING_TYPE_INFO;
	}
}
